package net.fruit.mq.client.impl.rocketmq;

import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import net.fruit.mq.client.Action;
import net.fruit.mq.client.Constants;
import net.fruit.mq.client.ConsumeContext;
import net.fruit.mq.client.Consumer;
import net.fruit.mq.client.Message;
import net.fruit.mq.client.MessageListener;
import net.fruit.mq.client.PropertyKeyConst;
import net.fruit.mq.client.PropertyValueConst;
import net.fruit.mq.client.exception.LMQClientException;
import net.fruit.mq.client.impl.tracehook.MQConsumeMessageHookImpl;
import net.fruit.mq.client.impl.util.ClientLoggerUtil;
import net.fruit.mq.client.trace.common.MQTraceConstants;
import net.fruit.mq.client.trace.dispatch.AsyncDispatcher;
import net.fruit.mq.client.trace.dispatch.impl.AsyncArrayDispatcher;

public class ConsumerImpl extends MQClientAbstract implements Consumer {
    private final static Logger log = ClientLoggerUtil.getClientLogger();
    private final DefaultMQPushConsumer defaultMQPushConsumer;
    private final ConcurrentHashMap<String/* Topic */, MessageListener> subscribeTable = new ConcurrentHashMap<String, MessageListener>();

    private final AtomicBoolean started = new AtomicBoolean(false);

    private AsyncDispatcher traceDispatcher=null;

    public ConsumerImpl(final Properties properties) {
        super(properties);
        this.defaultMQPushConsumer = new DefaultMQPushConsumer(new ClientRPCHook(sessionCredentials));

        String consumerGroup = properties.getProperty(PropertyKeyConst.ConsumerId);
        if (null == consumerGroup) {
            throw new LMQClientException("\'ConsumerId\' property is null");
        }

        String maxReconsumeTimes = properties.getProperty(PropertyKeyConst.MaxReconsumeTimes);
        if (!UtilAll.isBlank(maxReconsumeTimes)) {
            try {
                this.defaultMQPushConsumer.setMaxReconsumeTimes(Integer.parseInt(maxReconsumeTimes));
            } catch (NumberFormatException e) {
                //
            }
        }

        String consumeTimeout = properties.getProperty(PropertyKeyConst.ConsumeTimeout);
        if (!UtilAll.isBlank(consumeTimeout)) {
            try {
                this.defaultMQPushConsumer.setConsumeTimeout(Integer.parseInt(consumeTimeout));
            } catch (NumberFormatException e) {
                //
            }
        }

        boolean isVipChannelEnabled = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.isVipChannelEnabled, "false"));
        this.defaultMQPushConsumer.setVipChannelEnabled(isVipChannelEnabled);

        boolean postSubscriptionWhenPull = Boolean.parseBoolean(properties.getProperty(PropertyKeyConst.PostSubscriptionWhenPull, "false"));
        this.defaultMQPushConsumer.setPostSubscriptionWhenPull(postSubscriptionWhenPull);

        String messageModel = properties.getProperty(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
        this.defaultMQPushConsumer.setMessageModel(MessageModel.valueOf(messageModel));
        this.defaultMQPushConsumer.setConsumerGroup(consumerGroup);
        this.defaultMQPushConsumer.setInstanceName(this.buildIntanceName());
        this.defaultMQPushConsumer.setNamesrvAddr(this.getNameServerAddr());

        if (properties.containsKey(PropertyKeyConst.ConsumeThreadNums)) {
            this.defaultMQPushConsumer.setConsumeThreadMin(Integer.valueOf(properties.get(PropertyKeyConst.ConsumeThreadNums).toString()));
            this.defaultMQPushConsumer.setConsumeThreadMax(Integer.valueOf(properties.get(PropertyKeyConst.ConsumeThreadNums).toString()));
        }

        // 为Consumer增加消息轨迹回发模块
        try {
            Properties tempProperties = new Properties();
            tempProperties.put(MQTraceConstants.AccessKey, sessionCredentials.getAccessKey());
            tempProperties.put(MQTraceConstants.SecretKey, sessionCredentials.getSecretKey());
            tempProperties.put(MQTraceConstants.MaxMsgSize,"128000");
            tempProperties.put(MQTraceConstants.AsyncBufferSize,"2048");
            tempProperties.put(MQTraceConstants.MaxBatchNum, "100");
            tempProperties.put(MQTraceConstants.NAMESRV_ADDR,this.getNameServerAddr());
            tempProperties.put(MQTraceConstants.InstanceName,this.buildIntanceName());
            traceDispatcher=new AsyncArrayDispatcher(tempProperties);
            traceDispatcher.start(null,this.defaultMQPushConsumer.getInstanceName());
            this.defaultMQPushConsumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(
                    new MQConsumeMessageHookImpl(traceDispatcher));
        } catch (Throwable e) {
            log.error("system mqtrace hook init failed ,maybe can't send msg trace data");
        }
    }


    @Override
    public void start() {
        this.defaultMQPushConsumer.registerMessageListener(new MessageListenerImpl());
        try {
            if (this.started.compareAndSet(false, true)) {
                this.defaultMQPushConsumer.start();
            }
        } catch (Exception e) {
            throw new LMQClientException(e.getMessage());
        }
        if(this.traceDispatcher!=null){
            this.traceDispatcher.registerShutDownHook();
        }
    }


    @Override
    public void shutdown() {
        if (this.started.compareAndSet(true, false)) {
            this.defaultMQPushConsumer.shutdown();
        }
    }

    @Override
    public void subscribe(String topic, String subExpression, MessageListener listener) {
        if (null == topic) {
            throw new LMQClientException("topic is null");
        }

        if (null == listener) {
            throw new LMQClientException("listener is null");
        }

        try {
            this.subscribeTable.put(topic, listener);
            this.defaultMQPushConsumer.subscribe(topic, subExpression);
        } catch (MQClientException e) {
            throw new LMQClientException("defaultMQPushConsumer subscribe exception", e);
        }
    }

    @Override
    public void unsubscribe(String topic) {
        if (null != topic) {
            this.defaultMQPushConsumer.unsubscribe(topic);
        }
    }

    class MessageListenerImpl implements MessageListenerConcurrently {

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgsRMQList, ConsumeConcurrentlyContext contextRMQ) {
            MessageExt msgRMQ = msgsRMQList.get(0);
            Message msg = RMQUtil.msgConvert(msgRMQ);
            Map<String, String> stringStringMap = msgRMQ.getProperties();
            msg.setMsgID(msgRMQ.getMsgId());
            if (stringStringMap != null && stringStringMap.get(Constants.TRANSACTION_ID) != null) {
                msg.setMsgID(stringStringMap.get(Constants.TRANSACTION_ID));
            }
            MessageListener listener = ConsumerImpl.this.subscribeTable.get(msg.getTopic());
            if (null == listener) {
                throw new LMQClientException("MessageListener is null");
            }

            final ConsumeContext context = new ConsumeContext();
            Action action = listener.consume(msg, context);
            if (action != null) {
                switch (action) {
                    case CommitMessage:
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    case ReconsumeLater:
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    default:
                        break;
                }
            }

            return null;
        }
    }

    @Override
    public boolean isStarted() {
        return started.get();
    }

    @Override
    public boolean isClosed() {
        return !isStarted();
    }

}
